package io.wen.bd.s6m2.spark

import org.apache.spark.{SparkConf, SparkContext}

object ClickImpApp {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[*]").setAppName(this.getClass.getCanonicalName)

    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")

    val clickImpPath = "hdfs://node1:9000/data/spark/lg/s6m2/clickimp"
    val resultPath = "hdfs://node1:9000/data/spark/lg/s6m2/clickimpResult"

    val clickImpRDD = sc
      .textFile(clickImpPath, 1)
      .map(line => {
        // 解析行数据，提取广告id和日志类型
        val uri = line.split(" ")(3).split("\\?")
        val logType = if (uri(0).contains("click")) "click" else "imp"
        val params = uri(1).split("&")
        val adid = params(4).split("=")(1)
        (adid, logType)
      }).aggregateByKey((0, 0))(
      // 按adid聚合
      // 分区内合并，对click和imp计数
      (seqAcc, seqValue) => {
        var (clickCount, impCount) = (0, 0)
        if ("click".equals(seqValue)) {
          clickCount += 1
        } else if ("imp".equals(seqValue)) {
          impCount += 1
        }
        (seqAcc._1 + clickCount, seqAcc._2 + impCount)
      },
      // 分区间合并，累加click和imp数量
      (combAcc, combValue) => {
        (combAcc._1 + combValue._1, combAcc._2 + combValue._2)
      })
      .map(item => Array(item._1, item._2._1, item._2._2).mkString(","))

    clickImpRDD.saveAsTextFile(resultPath)
    clickImpRDD.foreach(println)

    Thread.sleep(1000000)
  }
}
